package ckR

//
//import (
//	"fmt"
//	"db2s/global"
//	"db2s/scheduleTask"
//	mq "db2s/topic-mq"
//	"runtime"
//	"strings"
//	"sync"
//	"sync/atomic"
//	"time"
//)
//
//// sampIsSegmentQueryListeners IsSegmentQuery  监听执行器
//func sampIsSegmentQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName         = string(message.Body)
//		plan              = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		logseq            = message.Properties["logSeq"].(int64)
//		sqlWhere          chan any
//		topic             = IndexConcurrentQuery
//		dTopic            = IsSegmentQuery
//		event             = "[sampIsSegmentQueryListeners]"
//		wg                sync.WaitGroup
//		taskObjectNeglect bool
//		taskObject        = scheduleTask.TaskDistributionO{}
//		done              = make(chan struct{})
//		deleteMapInfo     = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	pc, _, _, _ := runtime.Caller(1)
//	callingFunc := runtime.FuncForPC(pc)
//	callFile, callLine := callingFunc.FileLine(pc)
//	callFunc := callingFunc.Name()
//	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
//	plan.subTaskBeginTime = time.Now().UnixNano()
//	if GlobalPConfigs.result.Teletypewriter == "bar" {
//		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "running")
//		global.Wlog.Debug(fmt.Sprintf("(%d) %v %s Add task Bar Run object %v successfully", logseq, callFuncInfo, event, tableName))
//	}
//	//统计原目标端行数
//	if !sampMissIndexCount(plan, logseq) && !GlobalPConfigs.rules.SampR.IgCount {
//		return mq.ConsumeSuccess
//	}
//	//计算抽样比例
//	plan.sampRatioRows()
//	//计算需要抽样的块数量
//	plan.sampChunkSum()
//	//设计令牌桶
//	plan.generalTokenBucket()
//	wg.Add(2)
//	//init status
//	go expensiveOperation(&wg, message, done, int64(len(plan.samp.PartitionData.SourceName)), logseq)
//	go mqTableStatusMonitor(done, &wg, plan, logseq)
//	//处理wheresql
//	if v, ok := GlobalPConfigs.schema.WhereSql[fmt.Sprintf("%s%s%s", plan.schema, global.SchemaTableSplist, plan.table)]; ok && v != "" {
//		sqlWhere = make(chan any, 1)
//		sqlWhere <- map[string]string{"whereSql": v}
//		close(sqlWhere)
//		taskObjectNeglect = true
//	} else {
//		for _, partitionName1 := range plan.samp.PartitionData.SourceName {
//			wg.Add(1)
//			go func(partitionName string) {
//				defer func() {
//					wg.Done()
//				}()
//				var labelChecklist = make(map[string][]string)
//				//var insertPrefix string
//				//var err error
//				idx1 := plan.idxc
//				idx1.PartitionName = partitionName
//				if !strings.EqualFold(partitionName, "single") && len(partitionName) > 0 {
//					idx1.ShardName = plan.samp.PartitionData.SourcePartitionShardMap[partitionName]
//					idx1.PartitionId = plan.samp.PartitionData.SourcePartitionShardIdMap[partitionName]
//				}
//				taskObject = scheduleTask.GetTaskDistributionO(map[string]any{
//					"cpLength":         GlobalPConfigs.rules.QueueSize,
//					"whereColumn":      plan.indexColumn,
//					"schema":           plan.schema,
//					"mschema":          plan.mschema,
//					"table":            plan.table,
//					"mtable":           plan.mtable,
//					"logSeq":           logseq,
//					"tableLimit":       plan.tableLimit,
//					"chunkStartSource": plan.chunkStartSource,
//					"sdb":              GlobalPConfigs.SDB,
//					"ddb":              GlobalPConfigs.DDB,
//					"chunkStartFs":     idx1,
//					"chunkStartFd":     plan.dstIdxc,
//					"chunkNumber":      GlobalPConfigs.rules.ChanRowCount,
//					"tableColumn":      plan.tableColData.SColumnInfo,
//				})
//				sqlWhere = taskObject.TmpIsSegmentQuery("", plan.indexColumn)
//				//send message
//				if !chanMonitorSendMsg(sqlWhere,
//					getCheckMod(), topic, tableName, sendMsg, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//						"taskObject": taskObject, "taskObjectNeglect": taskObjectNeglect,
//						"sendMsgKey": "segmentSql", "label": map[string]any{"no": []string{}},
//						"err": plan.errEnd, "subTask": plan.subTaskInfo, "topic": string(topic),
//						"outFay": GlobalPConfigs.result.Teletypewriter, "sourcePartitionName": partitionName}, labelChecklist, func() {
//					}) {
//					return
//				}
//				if atomic.LoadInt64(plan.errEnd) > -1 {
//					atomic.AddInt64(plan.mqProductCustomerMonitor.mqLevelProductSend, int64(len(labelChecklist)))
//					atomic.AddInt64(plan.mqProductCustomerMonitor.mqStartTiming, 1)
//				}
//			}(partitionName1)
//		}
//	}
//	wg.Wait()
//	return mq.ConsumeSuccess
//}
//
//// sampIndexConcurrentQueryListeners IndexConcurrentQuery 监听执行器
//func sampIndexConcurrentQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName         = string(message.Body)
//		logseq            = message.Properties["logSeq"].(int64)
//		plan              = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		topic             = QueryTableSql
//		dTopic            = IndexConcurrentQuery
//		event             = "[rowsIndexConcurrentQueryListeners]"
//		taskObject        = message.Properties["taskObject"]
//		taskObjectNeglect bool
//		fatherSonTags     = make(map[string][]string)
//		//uid               uuid.UUID
//		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter} //
//	)
//	defer deleteTopic(deleteMapInfo)
//	if e := atomic.LoadInt64(plan.errEnd); e < 0 {
//		return mq.ConsumeSuccess
//	}
//	segmentSql := message.Properties["segmentSql"].(global.IndexColumnStartPartValueP)
//	//分段查询错误
//	if len(segmentSql.Error) > 0 {
//		//plan.errorAction(message, logseq, segmentSql.Error)
//		return mq.ConsumeSuccess
//	}
//	//where sql
//	//if len(segmentSql.WhereSql) > 0 {
//	//	taskObjectNeglect = true
//	//	var uidmap = make(map[string]any)
//	//	for k := range message.Properties["label"].(map[string]any) {
//	//		uid = uuid.New()
//	//		uidmap[k] = uid.String()
//	//		fatherSonTags[k] = []string{uid.String()}
//	//	}
//	//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//	//		"sqlwhere": segmentSql, "err": plan.errEnd, "subTask": plan.subTaskInfo,
//	//		"topic": string(topic), "label": uidmap,
//	//		"outFay": GlobalPConfigs.result.Teletypewriter}) {
//	//		return mq.ReconsumeLater
//	//	}
//	//	if atomic.LoadInt64(plan.errEnd) > -1 {
//	//		plan.mqProductCustomerMonitor.ProductMessage <- fatherSonTags
//	//		atomic.AddInt64(plan.mqProductCustomerMonitor.mqLevelCustomerReceive, 1)
//	//	}
//	//}
//	//多并发处理
//	partitionName := message.Properties["sourcePartitionName"].(string)
//	plan.idxc.PartitionName = partitionName
//	if !strings.EqualFold(partitionName, "single") && len(partitionName) > 0 {
//		plan.idxc.ShardName = plan.samp.PartitionData.SourcePartitionShardMap[partitionName]
//		plan.idxc.PartitionId = plan.samp.PartitionData.SourcePartitionShardIdMap[partitionName]
//	}
//	taskObject = scheduleTask.GetTaskDistributionO(map[string]any{
//		"cpLength":         GlobalPConfigs.rules.QueueSize,
//		"whereColumn":      plan.indexColumn,
//		"schema":           plan.schema,
//		"mschema":          plan.mschema,
//		"table":            plan.table,
//		"mtable":           plan.mtable,
//		"logSeq":           logseq,
//		"tableLimit":       int64(GlobalPConfigs.rules.ChanRowCount),
//		"chunkStartSource": plan.chunkStartSource,
//		"sdb":              GlobalPConfigs.SDB,
//		"ddb":              GlobalPConfigs.DDB,
//		"chunkStartFs":     plan.idxc,
//		"chunkStartFd":     plan.dstIdxc,
//		"chunkNumber":      GlobalPConfigs.rules.ChanRowCount,
//		"tableColumn":      plan.tableColData.SColumnInfo,
//	})
//	if !chanMonitorSendMsg(taskObject.(scheduleTask.TaskDistributionO).TmpIndexConcurrentQuery(segmentSql),
//		getCheckMod(), topic, tableName, sendMsg, map[string]any{"logSeq": logseq,
//			"taskObject": taskObject, "taskObjectNeglect": taskObjectNeglect,
//			"sendMsgKey": "sqlwhere", "subTask": plan.subTaskInfo, "label": message.Properties["label"],
//			"plan": message.Properties["plan"], "sourcePartitionName": message.Properties["sourcePartitionName"],
//			//"insertPrefix": message.Properties["insertPrefix"],
//			"topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}, fatherSonTags, func() {
//		}) {
//		return mq.ReconsumeLater
//	}
//	if atomic.LoadInt64(plan.errEnd) > -1 {
//		plan.mqProductCustomerMonitor.ProductMessage <- fatherSonTags
//		atomic.AddInt64(plan.mqProductCustomerMonitor.mqLevelCustomerReceive, 1)
//	}
//	atomic.AddUint64(plan.samp.CheckRows.ChunkSqlSum, uint64(len(fatherSonTags)))
//	return mq.ConsumeSuccess
//}
//
//// sampQueryTableSqlListeners rowsQueryTableSql 监听执行器
//func sampQueryTableSqlListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName = string(message.Body)
//		//sqlWhere  map[string]string
//		plan              = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		logseq            = message.Properties["logSeq"].(int64)
//		topic             = QueryTableData
//		dTopic            = QueryTableSql
//		event             = "[rowsQueryTableSqlListeners]"
//		taskObjectNeglect bool
//		taskObject        = message.Properties["taskObject"]
//		deleteMapInfo     = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	if e := atomic.LoadInt64(plan.errEnd); e < 0 {
//		plan.quit()
//		return mq.ConsumeSuccess
//	}
//	//if !plan.sendDatafixMqMsgStatus() {
//	//	return mq.ConsumeSuccess
//	//}
//	sqlWhere := message.Properties["sqlwhere"].(global.IndexColumnStartPartValueP)
//	//分段查询错误
//	if len(sqlWhere.Error) > 0 {
//		//plan.errorAction(message, logseq, sqlWhere.Error)
//		plan.quit()
//		return mq.ConsumeSuccess
//	}
//	partitionName := message.Properties["sourcePartitionName"].(string)
//	token, ok := <-plan.samp.Bucket.TokenBucket
//	if !ok {
//		plan.quit()
//		return mq.ConsumeSuccess
//	}
//	if _, ok1 := plan.samp.Bucket.RandomValue[token]; !ok1 {
//		plan.quit()
//		return mq.ConsumeSuccess
//	}
//	atomic.AddInt64(plan.subTaskInfo.Samp.SampProportionChunk, -1)
//	tableSql, err := plan.sampQueryTableSql(sqlWhere, partitionName)
//	if err != nil {
//		plan.quit()
//		return mq.ConsumeSuccess
//	}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"taskObject": taskObject, "taskObjectNeglect": taskObjectNeglect, "label": message.Properties["label"],
//		"err": plan.errEnd, "strsql": tableSql, "sourcePartitionName": message.Properties["sourcePartitionName"],
//		"subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// sampQueryTableDataSqlListeners QueryTableData 监听执行器
//func sampQueryTableDataSqlListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		querySql          map[string]*string
//		stt, dtt          any
//		err               error
//		logseq            = message.Properties["logSeq"].(int64)
//		tableName         = string(message.Body)
//		plan              = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		event             = "[rowsQueryTableDataSqlListeners]"
//		topic             = Md5Rows
//		dTopic            = QueryTableData
//		taskObject        = message.Properties["taskObject"]
//		taskObjectNeglect = message.Properties["taskObjectNeglect"].(bool)
//		deleteMapInfo     = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	//defer plan.quit()
//	if e := atomic.LoadInt64(plan.errEnd); e < 0 {
//		plan.quit()
//		return mq.ConsumeSuccess
//	}
//	tableSql := message.Properties["strsql"].(map[string]*string)
//	//tableSqlCrc32 := message.Properties["strsqlCrc32"].(string)
//	switch GlobalPConfigs.rules.RowsR.Options {
//	case "crc32", "md5":
//		//querySql = tableSqlCrc32
//	case "data":
//		querySql = tableSql
//	default:
//		querySql = tableSql
//	}
//	if stt, dtt, err = plan.sampQueryTableData(querySql, logseq); err != nil {
//		//plan.errorAction(message, logseq, err)
//		plan.quit()
//		return mq.ConsumeSuccess
//	}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"taskObject": taskObject, "taskObjectNeglect": taskObjectNeglect, "label": message.Properties["label"],
//		"sourceD": stt, "destD": dtt, "subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		plan.quit()
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// sampMd5Listeners rowsMd5 监听执行器
//func sampMd5Listeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName         = string(message.Body)
//		logseq            = message.Properties["logSeq"].(int64)
//		plan              = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		topic             = Arrcmp
//		dTopic            = Md5Rows
//		event             = "[rowsMd5Listeners]"
//		taskObject        = message.Properties["taskObject"]
//		taskObjectNeglect = message.Properties["taskObjectNeglect"].(bool)
//		deleteMapInfo     = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	if !plan.sendDatafixMqMsgStatus() {
//		plan.quit()
//		return mq.ReconsumeLater
//	}
//	stt, dtt := message.Properties["sourceD"].([]*string), message.Properties["destD"].([]*string)
//	stt1, dtt1, abs := plan.TmpMd5Rows(stt, dtt, logseq)
//	if !abs {
//		plan.quit()
//		return mq.ConsumeSuccess
//	}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"taskObject": taskObject, "taskObjectNeglect": taskObjectNeglect, "label": message.Properties["label"],
//		"sourceD": stt1, "destD": dtt1, "subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		plan.quit()
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// sampArrcmpListeners rowsArrcmp 监听执行器
//func sampArrcmpListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName         = string(message.Body)
//		logseq            = message.Properties["logSeq"].(int64)
//		plan              = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		topic             = AbnormalDataDiff
//		dTopic            = Arrcmp
//		event             = ""
//		taskObject        = message.Properties["taskObject"]
//		taskObjectNeglect = message.Properties["taskObjectNeglect"].(bool)
//		deleteMapInfo     = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	stt, dtt := message.Properties["sourceD"], message.Properties["destD"]
//	dataDisPos := plan.sampArrcmp(stt, dtt, logseq)
//	if len(dataDisPos) == 0 {
//		plan.quit()
//		return mq.ConsumeSuccess
//	}
//
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"taskObject": taskObject, "taskObjectNeglect": taskObjectNeglect, "label": message.Properties["label"],
//		"dataDisPos": dataDisPos, "subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		plan.quit()
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// sampAbnormalDataDiffListeners AbnormalDataDiff 监听执行器
//func sampAbnormalDataDiffListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName         = string(message.Body)
//		logseq            = message.Properties["logSeq"].(int64)
//		plan              = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		topic             = DataFixDiff
//		dTopic            = AbnormalDataDiff
//		event             = "[rowsAbnormalDataDiffListeners]"
//		taskObject        = message.Properties["taskObject"]
//		taskObjectNeglect = message.Properties["taskObjectNeglect"].(bool)
//		deleteMapInfo     = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	//if !plan.sendDatafixMqMsgStatus() {
//	//	atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
//	//	return mq.ReconsumeLater
//	//}
//	dataDisPos := message.Properties["dataDisPos"].(map[string][]*string)
//	diff, err := plan.sampAbnormalDataDiff(dataDisPos, logseq)
//	if err != nil {
//		plan.quit()
//		//plan.errorAction(message, logseq, err)
//		return mq.ConsumeSuccess
//	}
//
//	//if diff == "" {
//	//	atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
//	//	return mq.ReconsumeLater
//	//}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"taskObject": taskObject, "taskObjectNeglect": taskObjectNeglect, "label": message.Properties["label"],
//		"diffSql": diff, "subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		plan.quit()
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// rowsDataFixDiffListeners DataFixDiff 监听执行器
//func sampDataFixDiffListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		logseq        = message.Properties["logSeq"].(int64)
//		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		dTopic        = DataFixDiff
//		event         string
//		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	defer plan.quit()
//	//if !plan.sendDatafixMqMsgStatus() {
//	//	return mq.ReconsumeLater
//	//}
//	diffSql := message.Properties["diffSql"].(map[Logo][]*string)
//	if !strings.EqualFold(GlobalPConfigs.repair.DataFix, "no") {
//		if err := plan.sampDataFixDiff(diffSql, logseq); err != nil {
//			//plan.errorAction(message, logseq, err)
//			return mq.ConsumeSuccess
//		}
//	}
//	return mq.ConsumeSuccess
//}
